package rocketmq

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"github.com/star-find-cloud/star-chat/conf"
	log "github.com/star-find-cloud/star-chat/pkg/logger"
	"sync"
)

type EventProducer struct {
	Producer rocketmq.Producer
	Topic    string
}

var (
	producerMutex   = &sync.Mutex{}
	txProducerMutex = &sync.Mutex{}
)

func NewProducer(topic string) (*EventProducer, error) {
	producerMutex.Lock()
	defer producerMutex.Unlock()

	var config = conf.GetConfig()
	group := fmt.Sprintf("%s_%s_producer_group", config.App.Name, topic)

	mqProducer, err := rocketmq.NewProducer(
		producer.WithCreateTopicKey(topic),
		producer.WithNameServer([]string{config.MQ.RocketMQ.NameServer}),
		producer.WithRetry(3),
		producer.WithGroupName(group))
	if err != nil {
		log.AppLogger.Errorf("init producer error: %s", err)
		return nil, err
	}

	if mqProducer == nil {
		log.MQLogger.Errorf("producer is not initialized")
		return nil, fmt.Errorf("producer is not initialized")
	}

	err = mqProducer.Start()
	if err != nil {
		log.AppLogger.Errorf("start producer error: %s", err)
		return nil, err
	}
	eventProducer := &EventProducer{
		Producer: mqProducer,
		Topic:    topic,
	}

	log.MQLogger.Infof("init producer success, topic: %s", topic)
	return eventProducer, err
}

// SendSyncMsg 同步发送消息 	body:消息体 tags:消息标签(用于指定操作) keys:消息唯一标识
func (p *EventProducer) SendSyncMsg(ctx context.Context, body interface{}, tags, keys string) error {
	errChan := make(chan error, 1)
	defer close(errChan)

	go func() {
		jsonMsg, _ := json.Marshal(body)
		msg := primitive.NewMessage(p.Topic, jsonMsg)
		if tags != "" {
			msg.WithTag(tags)
		}
		if keys != "" {
			msg.WithKeys([]string{keys})
		}

		_, err := p.Producer.SendSync(ctx, msg)
		if err != nil {
			log.AppLogger.Errorf("send message error: %s", err)
			errChan <- err
		}

		log.MQLogger.Infoln("send message success")
		errChan <- nil
	}()

	if err := <-errChan; err != nil {
		return fmt.Errorf("send message error: %s", err)
	}

	return nil
}

// SendAsyncMsg 异步发送消息,  body:消息体 tags:消息标签(用于指定操作) keys:消息唯一标识
func (p *EventProducer) SendAsyncMsg(ctx context.Context, body interface{}, tags, keys string) error {
	errChan := make(chan error, 1)
	defer close(errChan)

	go func() {
		jsonMsg, _ := json.Marshal(body)
		msg := primitive.NewMessage(p.Topic, jsonMsg)
		if tags != "" {
			msg.WithTag(tags)
		}
		if keys != "" {
			msg.WithKeys([]string{keys})
		}

		err := p.Producer.SendAsync(ctx, func(ctx context.Context, result *primitive.SendResult, err error) {
			if err != nil {
				log.MQLogger.Errorf("send message failed: %v", err)
			} else {
				log.MQLogger.Infof("Async send success! MsgID: %s, Status: %v\n", result.MsgID, result.Status)
			}
		}, msg)

		if err != nil {
			log.AppLogger.Errorf("send message error: %v", err)
			errChan <- err
		}

		log.MQLogger.Infoln("send message success")
		errChan <- nil
	}()

	if err := <-errChan; err != nil {
		return fmt.Errorf("send message error: %v", err)
	}

	return nil
}

// SendOneWayMsg 单向发送, body:消息体 tags:消息标签(用于指定操作) keys:消息唯一标识
func (p *EventProducer) SendOneWayMsg(ctx context.Context, body interface{}, tags, keys string) error {
	errChan := make(chan error, 1)
	defer close(errChan)

	go func() {
		jsonMsg, _ := json.Marshal(body)
		msg := primitive.NewMessage(p.Topic, jsonMsg)
		if tags != "" {
			msg.WithTag(tags)
		}
		if keys != "" {
			msg.WithKeys([]string{keys})
		}

		err := p.Producer.SendOneWay(ctx, msg)
		if err != nil {
			log.AppLogger.Errorf("send message error: %s", err)
			errChan <- err
		}

		log.MQLogger.Infoln("send message success")
		errChan <- nil
	}()

	if err := <-errChan; err != nil {
		return fmt.Errorf("send message error: %s", err)
	}
	return nil
}

func (p *EventProducer) Shutdown() error {
	if p.Producer != nil {
		return p.Producer.Shutdown()
	}
	return nil
}

// TransactionProducer 事务消息结构体
type TransactionProducer struct {
	Producer rocketmq.TransactionProducer
	Topic    string
}

// NewTXProducer 初始化事务消息结构体
func NewTXProducer(topic string, listener primitive.TransactionListener) (*TransactionProducer, error) {
	txProducerMutex.Lock()
	defer txProducerMutex.Unlock()
	var config = conf.GetConfig()
	group := fmt.Sprintf("%s_%s_tx_producer_group", config.App.Name, topic)

	txProducer, err := rocketmq.NewTransactionProducer(listener,
		producer.WithCreateTopicKey(topic),
		producer.WithNameServer([]string{config.MQ.RocketMQ.NameServer}),
		producer.WithRetry(3),
		producer.WithGroupName(group),
	)
	if err != nil {
		log.AppLogger.Errorf("init tx producer error: %s", err)
		return nil, err
	}

	if txProducer == nil {
		log.MQLogger.Errorf("tx producer is not initialized")
		return nil, fmt.Errorf("tx producer is not initialized")
	}

	err = txProducer.Start()
	if err != nil {
		log.AppLogger.Errorf("start tx producer error: %s", err)
		return nil, err
	}

	log.MQLogger.Infof("init tx producer success, topic: %s", topic)

	return &TransactionProducer{Producer: txProducer, Topic: topic}, nil
}

func (p *TransactionProducer) SendMsg(ctx context.Context, body interface{}, tags string, keys []string) error {
	if p.Producer == nil {
		log.MQLogger.Errorf("tx producer is not initialized")
		return fmt.Errorf("tx producer is not initialized")
	}

	jsonMsg, _ := json.Marshal(body)
	msg := primitive.NewMessage(p.Topic, jsonMsg)
	if tags != "" {
		msg.WithTag(tags)
	}
	if keys != nil {
		msg.WithKeys(keys)
	}

	result, err := p.Producer.SendMessageInTransaction(ctx, msg)
	if err != nil {
		log.AppLogger.Errorf("send tx message error: %s", err)
		return err
	}

	if result.Status != primitive.SendOK {
		log.MQLogger.Errorf("send tx message error: %v", result.Status)
		return fmt.Errorf("send tx message error: %v", result.Status)
	}

	switch result.State {
	case primitive.RollbackMessageState:
		log.MQLogger.Warnf("send tx message rollback, msgID: %s", result.MsgID)
	case primitive.UnknowState:
		log.MQLogger.Warnf("send tx message unknow, msgID: %s", result.MsgID)
	default:
		log.MQLogger.Infof("send tx message success, msgID: %s", result.MsgID)
	}

	log.MQLogger.Infof("send tx message success, msgID: %s", result.MsgID)

	return nil
}
